package com.shujia.core

import java.util
import java.util.Properties

import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

object Demo3CanalJson {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "master:9092")
    properties.setProperty("group.id", "asdasdsa")

    //创建flink kafka 消费者
    val flinkKafkaConsumer = new FlinkKafkaConsumer[String]("student.order", new SimpleStringSchema(), properties)


    flinkKafkaConsumer.setStartFromLatest()


    val canalDS: DataStream[String] = env.addSource(flinkKafkaConsumer)


    val orderDS: DataStream[(String, Double)] = canalDS.map(line => {

      //将一个json的字符串转换成json的对象
      //json的对象可以通过key获取value
      val jsonObj: JSONObject = JSON.parseObject(line)

      val datas: JSONArray = jsonObj.getJSONArray("data")

      val data: JSONObject = datas.getJSONObject(0)


      val id: String = data.getString("id")
      val order_id: String = data.getString("order_id")
      val amount: Double = data.getString("amount").toDouble
      val create_time: String = data.getString("create_time")


      //获取类型
      val t: String = jsonObj.getString("type")

      var acc = 0.0


      t match {
        case "UPDATE" =>
          val oldJson: JSONObject = jsonObj.getJSONArray("old").getJSONObject(0)
          val oldCmount: Double = oldJson.getString("amount").toDouble
          acc = amount - oldCmount

        case "DELETE" =>
          acc = -amount

        case "INSERT" =>
          acc = amount
      }

      (id, acc)
    })

    orderDS
      .keyBy(_._1)
      .sum(1)
      .print()


    env.execute()


  }

}
